/*
* Copyright (C) 2020 ~ 2024 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkRtpTcpServer.h"
#include "MkUtil/MkLog.h"

CMkRtpTcpSession::CMkRtpTcpSession(CMkEvent* pEvent, CMkTimer* pTimer, CMkTcpServer* pServer)
    : CMkTcpSession(pEvent, pTimer, pServer)
    , m_nMaxLen(Len2K)
    , m_nOffset(0)
{
    m_pRecvBuffer = (Uint8*)malloc(m_nMaxLen);
}


Uint32 CMkRtpTcpSession::OnRecv()
{
    Uint32 nToRead = m_nMaxLen - m_nOffset;
    if (0 == nToRead) {
        m_nOffset = 0;
        nToRead = m_nMaxLen;
        MkWarningLog("clear not parse sucess buffer\n");
    }
    Uint32 ErrorCode = Recv(m_pRecvBuffer + m_nOffset, nToRead);
    MkCheckErrorLog(ErrorCode, "rtp tcp client recv error\n");
    MkInfoLog("recv rtp tcp session msg %s:%d\n", GetRemoteHost().c_str(), GetRemotePort());
    CMkRtpTcpServer *pServer = dynamic_cast<CMkRtpTcpServer*>(GetParentServer());
    if (!pServer) {
        return UnknownError;
    }

    m_nOffset += nToRead;
    Uint32 nUsed = 0;
    Uint16 PacketLen = 0;
    while (nUsed + 2 < m_nOffset) {
        PacketLen = (m_pRecvBuffer[nUsed] << 8) | (m_pRecvBuffer[nUsed + 1]);
        if (nUsed + 2 + PacketLen > m_nOffset) {
            break;
        }
        nUsed += 2;
        pServer->ParsePacket(m_pRecvBuffer + nUsed, PacketLen, GetRemoteHost(), GetRemotePort(), this);
        nUsed += PacketLen;
    }
    memmove(m_pRecvBuffer, m_pRecvBuffer + nUsed, m_nOffset - nUsed);
    m_nOffset -= nUsed;
    return NoneError;
}

CMkRtpTcpServer::CMkRtpTcpServer(CMkEventPool* pEventPool, CMkTimer* pTimer)
    : CMkTcpServer(pEventPool, pTimer, 10000)
    , CMkRtp(pTimer)  //rtp over tcp not check rtcp
{

}

CMkTcpSession* CMkRtpTcpServer::CreateNewSession(CMkEvent *pEvent, CMkTimer* pTimer)
{
    return new CMkRtpTcpSession(pEvent, pTimer, this);
}

Uint32 CMkRtpTcpServer::OnSessionAccepted(CMkTcpSession *pSession)
{
    //if remote not the sdp remoter host  remote port  client mast send rtcp with ssrc first
    CMkRtpTcpSession* pRtpSession = dynamic_cast<CMkRtpTcpSession*>(pSession);
    Uint32 Ssrc = GetRemoterSsrc(pSession->GetRemoteHost(), pSession->GetRemotePort());
    if (0 != Ssrc && pRtpSession) {
        pRtpSession->SetSsrc(Ssrc);
        m_RtpSessionLock.WLock();
        m_MapRtpSession[Ssrc] = dynamic_cast<CMkRtpTcpSession*>(pSession);
        m_RtpSessionLock.WUnlock();
        MkRtpRemoter Remoter;
        GetRemoter(Ssrc, Remoter);
        if (MkRtpRecver & Remoter.RtpType) {
            CMkRtcpPacket RecverReportRtcp;
            RecverReportRtcp.SetPayload(MkRtcpRr);
            RecverReportRtcp.SetReceptionSsrc(Remoter.Ssrc);
            SendRtcpPacket(RecverReportRtcp, Remoter.RtpRemoteHost, Remoter.RtpRemotePort + 1);
        }
        if (MkRtpSender & Remoter.RtpType) {
            CMkRtcpPacket SenderReportRtcp;
            SenderReportRtcp.SetPayload(MkRtcpSr);
            SenderReportRtcp.SetSenderReportSsrc(Remoter.Ssrc);
            SenderReportRtcp.SetSenderReportNtpTimeStamp(NowMkTime.GetTvSec());
            SenderReportRtcp.SetSenderReportRtpTimeStamp(Remoter.TimeStamp);
            SenderReportRtcp.SetSenderReportSenderPacketCount(Remoter.SendPacketCount);
            SenderReportRtcp.SetSenderReportSenderOctetCount(0);
            SendRtcpPacket(SenderReportRtcp, Remoter.RtpRemoteHost, Remoter.RtpRemotePort + 1);
        }

        CMkRtcpPacket DescRtcp;
        DescRtcp.SetPayload(MkRtcpSdes);
        DescRtcp.SetSourceDescriptionSsrc(Remoter.Ssrc);
        DescRtcp.SetSourceDescriptionAddChunk(MkSdesName, "MkRtpServer");
        SendRtcpPacket(DescRtcp, Remoter.RtpRemoteHost, Remoter.RtpRemotePort + 1);
    }
    return NoneError;
}

Uint32 CMkRtpTcpServer::OnSessionEventRemoved(CMkTcpSession* pSession)
{
    CMkRtpTcpSession* pRtpSession = dynamic_cast<CMkRtpTcpSession*>(pSession);
    if (pRtpSession) {
        m_RtpSessionLock.WLock();
        MkUnorderedMap<Uint32, CMkRtpTcpSession*>::iterator it = m_MapRtpSession.find(pRtpSession->GetSsrc());
        if (it != m_MapRtpSession.end()) {
            m_MapRtpSession.erase(it);
        }
        m_RtpSessionLock.WUnlock();
        MkRtpRemoter Remoter;
        GetRemoter(pRtpSession->GetSsrc(), Remoter);
        if (Remoter.RtpClosedCb) {
            Remoter.RtpClosedCb();
        }
    }

    return CMkTcpServer::OnSessionEventRemoved(pSession);
}

Uint32 CMkRtpTcpServer::Release()
{
    //do nothing
    return NoneError;
}

Uint32 CMkRtpTcpServer::ParsePacket(Uint8* pBuf, Uint32 nLen, const MkString& SenderHost, const Uint16& SenderPort, CMkRtpTcpSession* pRtpSession)
{
    Uint32 ErrorCode = NoneError;
    CMkRtpPacket RtpPacket;
    CMkRtcpPacket RtcpPacket;
    if (NoneError == RtpPacket.Parse(pBuf, nLen)) {
        ParseRtpPacket(RtpPacket, SenderHost, SenderPort);
    } else {
        Uint32 nUsed = 0;
        Uint32 nCurUsed = 0;
        while (nUsed < nLen) {
            ErrorCode = RtcpPacket.Parse(pBuf + nUsed, nLen - nUsed, nCurUsed);
            if (NoneError != ErrorCode) {
                break;
            }
            ParseRtcpPacket(RtcpPacket, SenderHost, SenderPort);
            Uint32 Ssrc = 0;
            switch (RtcpPacket.GetPayload()) {
            case MkRtcpSr:
                Ssrc = RtcpPacket.GetSenderReportSsrc();
                break;
            case MkRtcpRr:
                Ssrc = RtcpPacket.GetReceptionSsrc();
                break;
            case MkRtcpSdes:
                Ssrc = RtcpPacket.GetSourceDescriptionSsrc();
                break;
            case MkRtcpApp:
                Ssrc = RtcpPacket.GetAppSsrc();
                break;
            default:
                break;
            }
            if (0 != Ssrc) {
                pRtpSession->SetSsrc(Ssrc);
                m_RtpSessionLock.WLock();
                m_MapRtpSession[Ssrc] = pRtpSession;
                m_RtpSessionLock.WUnlock();
            }
            nUsed += nCurUsed;
        }
    }
    return NoneError;
}

Uint32 CMkRtpTcpServer::SendRtpPacket(const CMkRtpPacket& RtpPacket, const MkString& RemoteIp, Uint16 RemotePort)
{
    CMkRtpTcpSession *pSession = nullptr;
    m_RtpSessionLock.RLock();
    MkUnorderedMap<Uint32, CMkRtpTcpSession*>::iterator it = m_MapRtpSession.find(RtpPacket.GetSsrc());
    if (it != m_MapRtpSession.end()) {
        pSession = it->second;
    }
    m_RtpSessionLock.RUnlock();
    if (pSession) {
        CMkBufferList BufList;
        Uint16 nLen = RtpPacket.GetHeaderLen() + RtpPacket.GetPayloadLength();
        nLen = ntohs(nLen);
        BufList.Append(&nLen, sizeof(Uint16));
        BufList.Append(RtpPacket.GetHeaderData(), RtpPacket.GetHeaderLen());
        if (RtpPacket.GetPayloadData()) {
            BufList.Append(RtpPacket.GetPayloadData(), RtpPacket.GetPayloadLength());
        } else {
            BufList.Append(RtpPacket.GetPayloadBufferList());
        }
        MkInfoLog("recv rtp tcp session msg %s:%d\n", pSession->GetRemoteHost().c_str(), pSession->GetRemotePort());
        return pSession->Send(BufList);
    }
    return InvalidParams;
}

Uint32 CMkRtpTcpServer::SendRtcpPacket(const CMkRtcpPacket& RtcpPacket, const MkString& RemoteIp, Uint16 RemotePort)
{
    MkUnUse(RemotePort);
    MkUnUse(RemoteIp);
    CMkRtpTcpSession *pSession = nullptr;
    m_RtpSessionLock.RLock();
    MkUnorderedMap<Uint32, CMkRtpTcpSession*>::iterator it = m_MapRtpSession.find(RtcpPacket.GetSsrc());
    if (it != m_MapRtpSession.end()) {
        pSession = it->second;
    }
    m_RtpSessionLock.RUnlock();
    if (pSession) {
        Uint8 pBuf[Len1K] = { 0 };
        Uint32 nUsed = 0;
        RtcpPacket.Packet(pBuf, Len1K, nUsed);
        Uint16 nLen = nUsed;
        nLen = ntohs(nLen);
        CMkBufferList BufList;
        BufList.Append(&nLen, sizeof(Uint16));
        BufList.Append(pBuf, nUsed);
        MkInfoLog("recv rtcp tcp session msg %s:%d\n", pSession->GetRemoteHost().c_str(), pSession->GetRemotePort());
        pSession->Send(BufList);
    }
    return NoneError;
}

Uint32 CMkRtpTcpServer::DeleteRemoter(const Uint32& Ssrc)
{
    m_RtpSessionLock.WLock();
    MkUnorderedMap<Uint32, CMkRtpTcpSession*>::iterator it = m_MapRtpSession.find(Ssrc);
    if (it != m_MapRtpSession.end()) {
        it->second->Close();
        m_MapRtpSession.erase(it);
    }
    m_RtpSessionLock.WUnlock();
    return CMkRtp::DeleteRemoter(Ssrc);
}

CMkRtpTcpServer::~CMkRtpTcpServer()
{

}

CMkRtpTcpSession::~CMkRtpTcpSession()
{
    MkFree(m_pRecvBuffer);
}